#include "filters/filter_worker.h"
#include "utils/logger.h"

FilterThreadPool::FilterThreadPool(
    size_t threadCount, std::shared_ptr<IFilterEngine> filterEngine,
    std::shared_ptr<PacketDispatcher> dispatcher)
    : m_filterEngine(std::move(filterEngine)), m_dispatcher(std::move(dispatcher)), m_threadCount(threadCount) {
    createWorkers();
    LOG_INFO("FilterThreadPool created with " << threadCount << " workers");
}

FilterThreadPool::~FilterThreadPool() { stop(); }

void FilterThreadPool::createWorkers() {
    m_workers.reserve(m_threadCount);

    for (size_t i = 0; i < m_threadCount; ++i) {
        auto& queue = m_dispatcher->getQueue(i);
        auto worker = std::make_unique<FilterWorker>(i, m_filterEngine, queue);
        m_workers.emplace_back(std::move(worker));
    }
}

bool FilterThreadPool::start() {
    if (m_running.load(std::memory_order_acquire)) {
        LOG_WARNING("FilterThreadPool already running");
        return true;
    }

    bool allStarted = true;
    size_t startedCount = 0;
    for (auto&worker : m_workers) {
        if (!worker->start()) {
            LOG_ERROR("Failed to start filter worker " << worker->getId());
            allStarted = false;
        } else {
            startedCount++;
        }
    }

    m_running.store(allStarted, std::memory_order_release);
    if (allStarted) {
        LOG_INFO("FilterThreadPool started successfully with " << startedCount << " workers");
    } else {
        LOG_WARNING("FilterThreadPool started with errors: " << startedCount <<  "/" << m_workers.size() << " workers running");
    }
    return allStarted;
}

void FilterThreadPool::stop() {
    if (!m_running.load(std::memory_order_acquire)) return;

    for (auto&worker : m_workers) {
        worker->stop();
    }

    m_running.store(false, std::memory_order_release);
    LOG_INFO("FilterThreadPool stopped");
}

void FilterThreadPool::setOutputCallback(FilterWorker::PacketCallBack callback) {
    for (auto& worker : m_workers) {
        worker->setOutputCallback(callback);
    }
}

std::vector<FilterWorker::WorkerStats> FilterThreadPool::getAllStats() const {
    std::vector<FilterWorker::WorkerStats> stats;
    stats.reserve(m_workers.size());

    for (const auto& worker : m_workers) {
        stats.emplace_back(worker->getStats());
    }

    return stats;
}

size_t FilterThreadPool::getActiveThreadCount() const {
    size_t count = 0;
    for (const auto& worker : m_workers) {
        if (worker->isRunning()) {
            count++;
        }
    }
    return count;
}

FilterThreadPool::PoolStats FilterThreadPool::getPoolStats() const {
    PoolStats poolStats;
    poolStats.total_threads = m_workers.size();
    poolStats.active_threads = getActiveThreadCount();

    auto allStats = getAllStats();
    for (const auto& stats : allStats) {
        poolStats.total_processed_packets += stats.processed_packets;
        poolStats.total_accepted_packets += stats.accepted_packets;
        poolStats.total_dropped_packets += stats.dropped_packets;
    }

    /// Simplified design
    poolStats.overall_throughput_pps = poolStats.total_processed_packets;

    return poolStats;
}